package net.bwie.realtime.jtp.dwd.log.jjb;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.jtp.dwd.log.jjb
 * @Author: zhangleqing
 * @CreateTime: 2025-08-17  19:33
 * @Description: TODO 新老用户修复
 * @Version: 1.0
 */
public class Text_01 extends KeyedProcessFunction<String, String, String> {

    // 初始化一个容器，存储时间

    private transient ValueState<String> firstData;

    @Override
    public void open(Configuration parameters) throws Exception {
        firstData = getRuntimeContext().getState(
             new ValueStateDescriptor<String>("first-date", String.class)
        );
    }

    @Override
    public void processElement(String value,
                               Context ctx,
                               Collector<String> out) throws Exception {
        String firstDate = firstData.value();

        JSONObject jsonObject = JSON.parseObject(value);
        String is_new = jsonObject.getJSONObject("common").getString("is_new");
        Long ts = jsonObject.getLong("ts");

        // 条件判断
        if ("1".equals(is_new)){
            // 如果为空 填充当前时间
            if (null == firstDate){
                firstData.update(DateTimeUtil.convertLongToString(ts, DateTimeUtil.DATE_FORMAT));
            }else {
                if (!firstDate.equals(DateTimeUtil.convertLongToString(ts, DateTimeUtil.DATE_FORMAT))){
                    // 获取json字符串中的common中的is_new字段改成0
                    jsonObject.getJSONObject("common").put(is_new, 0);
                }
            }
        }
        else {
            // 填充昨天的时间  86400000毫秒=一天
            if (null == firstDate){
                firstData.update(DateTimeUtil.convertLongToString(ts-86400000, DateTimeUtil.DATE_FORMAT));
            }
        }

        out.collect(jsonObject.toJSONString());
    }


}
